/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2022/07/05.
//

#include "sql/expr/expression.h"
#include "common/log/log.h"
#include "common/value.h"
#include "sql/expr/tuple.h"
#include "sql/expr/arithmetic_operator.hpp"
#include "sql/operator/logical_operator.h"
#include "sql/operator/physical_operator.h"
#include "sql/operator/predicate_physical_operator.h"
#include "sql/parser/parse_defs.h"
#include "sql/stmt/select_stmt.h"
#include "storage/common/chunk.h"
#include "storage/common/column.h"
#include "storage/field/field.h"
#include <cstddef>
#include <functional>
#include <memory>
#include <cmath>
#include <string>
#include <vector>

using namespace std;

RC FieldExpr::get_value(const Tuple &tuple, Value &value) const
{
  // 如果是相关子查询，则从外层tuple中取得值
  if (is_correlated() && outside_tuples.size() > 0) {
    RC rc;
    for (unsigned i = 0; i < outside_tuples.size(); i++) {
      rc = outside_tuples[i]->find_cell(TupleCellSpec(table_name(), field_name()), value);
      if (rc == RC::SUCCESS) {
        return rc;
      }
    }
    return RC::SCHEMA_FIELD_MISSING;
  }
  return tuple.find_cell(TupleCellSpec(table_name(), field_name()), value);
}

bool FieldExpr::equal(const Expression &other) const
{
  if (this == &other) {
    return true;
  }
  if (other.type() != ExprType::FIELD) {
    return false;
  }
  const auto &other_field_expr = static_cast<const FieldExpr &>(other);
  return table_name() == other_field_expr.table_name() && field_name() == other_field_expr.field_name();
}

// TODO: 在进行表达式计算时，`chunk` 包含了所有列，因此可以通过 `field_id` 获取到对应列。
// 后续可以优化成在 `FieldExpr` 中存储 `chunk` 中某列的位置信息。
RC FieldExpr::get_column(Chunk &chunk, Column &column)
{
  if (pos_ != -1) {
    column.reference(chunk.column(pos_));
  } else {
    column.reference(chunk.column(field().meta()->field_id()));
  }
  return RC::SUCCESS;
}

bool ValueExpr::equal(const Expression &other) const
{
  if (this == &other) {
    return true;
  }
  if (other.type() != ExprType::VALUE) {
    return false;
  }
  const auto &other_value_expr = static_cast<const ValueExpr &>(other);
  return value_.compare(other_value_expr.get_value()) == 0;
}

RC ValueExpr::get_value(const Tuple &tuple, Value &value) const
{
  value = value_;
  return RC::SUCCESS;
}

RC ValueExpr::get_column(Chunk &chunk, Column &column)
{
  column.init(value_);
  return RC::SUCCESS;
}
//////////////////////////////////////////////////////////////////////////////////
///// TODO: 定义一些新的函数

SubQueryExpr::SubQueryExpr(SelectSqlNode &selection)
{
  sql_node_                        = std::make_unique<SelectSqlNode>();
  sql_node_->expressions           = std::move(selection.expressions);
  sql_node_->conditions            = std::move(selection.conditions);
  sql_node_->group_by              = std::move(selection.group_by);
  sql_node_->order_by              = std::move(selection.order_by);
  sql_node_->having                = std::move(selection.having);
  sql_node_->relations_and_aliases = selection.relations_and_aliases;
}

bool SubQueryExpr::equal(const Expression &other) const { return false; }

RC SubQueryExpr::get_value(const Tuple &tuple, Value &value) const
{
  value = value_;
  return RC::SUCCESS;
}

RC SubQueryExpr::get_column_values(vector<Value> &columns_values)
{

  // // 这里应该逐层往下找FieldExpr
  // if (cached_physical_operator_ != nullptr) {
  //   Tuple *tuple = cached_physical_operator_->current_tuple();
  //   LOG_TRACE("correlated sub query detected!");
  //   // 这里取得的应该是groupby physical operator
  //   vector<unique_ptr<PhysicalOperator>> &group_by_phy_opr = physical_operator_->children();
  //   if (group_by_phy_opr.empty()) {
  //     LOG_WARN("no group_by physical operator");
  //     return RC::INTERNAL;
  //   }
  //   vector<unique_ptr<PhysicalOperator>> &predicate_phy_opr = group_by_phy_opr[0]->children();
  //   if (predicate_phy_opr.empty()) {
  //     LOG_WARN("no predicate physical operator");
  //     return RC::INTERNAL;
  //   }
  //   PredicatePhysicalOperator *predicate_operator =
  //       static_cast<PredicatePhysicalOperator *>(predicate_phy_opr[0].get());
  //   unique_ptr<Expression> &conjunction_expr = predicate_operator->expression();
  //   // conjunction_expr 可能有多个ComparisonExpr
  //   auto expr_tmp = static_cast<ConjunctionExpr *>(conjunction_expr.get());
  //   for (auto &comp_expr : expr_tmp->children()) {
  //     ComparisonExpr *comparison_expr = static_cast<ComparisonExpr *>(comp_expr.get());
  //     if (comparison_expr->left()->type() == ExprType::FIELD) {
  //       FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->left().get());
  //       if (field_expr->is_correlated()) {
  //         field_expr->set_tuple(tuple);
  //       }
  //     }
  //     if (comparison_expr->right()->type() == ExprType::FIELD) {
  //       FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->right().get());
  //       if (field_expr->is_correlated()) {
  //         field_expr->set_tuple(tuple);
  //       }
  //     }
  //   }
  // }

  RC rc = RC::SUCCESS;
  if (physical_operator_ == nullptr) {
    LOG_WARN("sub query operator is null");
    return RC::INTERNAL;
  }
  rc = physical_operator_->open(nullptr);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to open sub query trx");
    return rc;
  }
  // 这里开始取得元组
  Tuple *tuple = nullptr;
  while (RC::SUCCESS == (rc = physical_operator_->next())) {
    tuple = physical_operator_->current_tuple();
    if (tuple == nullptr) {
      LOG_WARN("failed to get tuple from operator");
      physical_operator_->close();
      return RC::RECORD_EOF;
    }
    Value value;
    rc = tuple->cell_at(0, value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to find the cell in tuple!");
      rc = physical_operator_->close();
      return RC::INTERNAL;
    }
    columns_values.push_back(value);
  }
  physical_operator_->close();
  if (rc != RC::RECORD_EOF) {
    LOG_WARN("failed to get tuple from operator");
    return rc;
  }
  return RC::SUCCESS;
}

bool SubQueryExpr::has_tuple()
{
  if (physical_operator_ == nullptr) {
    LOG_WARN("sub query operator is null");
    return false;
  }
  RC rc = physical_operator_->open(nullptr);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to open sub query trx");
    return false;
  }
  bool   flag  = false;
  Tuple *tuple = nullptr;
  while (RC::SUCCESS == (rc = physical_operator_->next())) {
    tuple = physical_operator_->current_tuple();
    if (tuple != nullptr) {
      flag = true;
    }
  }
  physical_operator_->close();
  return flag;
}

RC SubQueryExpr::physical_get_value(Value &value) const
{

  // 这里应该逐层往下找FieldExpr
  if (cached_physical_operator_ != nullptr) {
    Tuple *tuple = cached_physical_operator_->current_tuple();
    LOG_TRACE("correlated sub query detected!");
    // 这里取得的应该是groupby physical operator
    vector<unique_ptr<PhysicalOperator>> &group_by_phy_opr = physical_operator_->children();
    if (group_by_phy_opr.empty()) {
      LOG_WARN("no group_by physical operator");
      return RC::INTERNAL;
    }
    vector<unique_ptr<PhysicalOperator>> &predicate_phy_opr = group_by_phy_opr[0]->children();
    if (predicate_phy_opr.empty()) {
      LOG_WARN("no predicate physical operator");
      return RC::INTERNAL;
    }
    PredicatePhysicalOperator *predicate_operator =
        static_cast<PredicatePhysicalOperator *>(predicate_phy_opr[0].get());
    unique_ptr<Expression> &conjunction_expr = predicate_operator->expression();
    // conjunction_expr 可能有多个ComparisonExpr
    auto expr_tmp = static_cast<ConjunctionExpr *>(conjunction_expr.get());
    for (auto &comp_expr : expr_tmp->children()) {
      ComparisonExpr *comparison_expr = static_cast<ComparisonExpr *>(comp_expr.get());
      if (comparison_expr->left()->type() == ExprType::FIELD) {
        FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->left().get());
        if (field_expr->is_correlated()) {
          field_expr->add_tuple(tuple);
        }
      }
      if (comparison_expr->right()->type() == ExprType::FIELD) {
        FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->right().get());
        if (field_expr->is_correlated()) {
          field_expr->add_tuple(tuple);
        }
      }
    }
  }

  // TODO: 这里使用物理逻辑算子得到value即可
  RC  rc  = RC::SUCCESS;
  int cnt = 0;  // 用来计数一共有多少行
  rc      = physical_operator_->open(nullptr);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to open sub query trx");
    return rc;
  }
  Tuple *tuple = nullptr;
  while (RC::SUCCESS == (rc = physical_operator_->next())) {
    cnt++;
    if (cnt == 1) {
      tuple = physical_operator_->current_tuple();
      rc    = tuple->cell_at(0, value);
      if (rc != RC::SUCCESS) {
        LOG_WARN("failed to find the cell in tuple!");
        rc = physical_operator_->close();
        return RC::INTERNAL;
      }
    }
  }
  if (cnt == 0 || tuple == nullptr) {
    LOG_WARN("sub query has no tuples");
    rc = physical_operator_->close();
    return RC::RECORD_EOF;
  }
  if (cnt > 1) {
    LOG_WARN("sub query has more than one tuple");
    rc = physical_operator_->close();
    return RC::INVALID_ARGUMENT;
  }
  if (tuple->cell_num() > 1) {
    LOG_WARN("tuple must has only one cell in sub query");
    rc = physical_operator_->close();
    return RC::INVALID_ARGUMENT;
  }
  // 这里表示数目都对上了
  rc = physical_operator_->close();
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to close sub query trx");
    return rc;
  }
  return RC::SUCCESS;
}

RC SubQueryExpr::get_column(Chunk &chunk, Column &column) { return RC::UNIMPLEMENTED; }

RC SubQueryExpr::try_get_value(Value &value) const
{
  // // 这里应该逐层往下找FieldExpr
  // if (cached_physical_operator_ != nullptr) {
  //   Tuple *tuple = cached_physical_operator_->current_tuple();
  //   LOG_TRACE("correlated sub query detected!");
  //   // 这里取得的应该是groupby physical operator
  //   vector<unique_ptr<PhysicalOperator>> &group_by_phy_opr = physical_operator_->children();
  //   if (group_by_phy_opr.empty()) {
  //     LOG_WARN("no group_by physical operator");
  //     return RC::INTERNAL;
  //   }
  //   vector<unique_ptr<PhysicalOperator>> &predicate_phy_opr = group_by_phy_opr[0]->children();
  //   if (predicate_phy_opr.empty()) {
  //     LOG_WARN("no predicate physical operator");
  //     return RC::INTERNAL;
  //   }
  //   PredicatePhysicalOperator *predicate_operator =
  //       static_cast<PredicatePhysicalOperator *>(predicate_phy_opr[0].get());
  //   unique_ptr<Expression> &conjunction_expr = predicate_operator->expression();
  //   // conjunction_expr 可能有多个ComparisonExpr
  //   auto expr_tmp = static_cast<ConjunctionExpr *>(conjunction_expr.get());
  //   for (auto &comp_expr : expr_tmp->children()) {
  //     ComparisonExpr *comparison_expr = static_cast<ComparisonExpr *>(comp_expr.get());
  //     if (comparison_expr->left()->type() == ExprType::FIELD) {
  //       FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->left().get());
  //       if (field_expr->is_correlated()) {
  //         field_expr->set_tuple(tuple);
  //       }
  //     }
  //     if (comparison_expr->right()->type() == ExprType::FIELD) {
  //       FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->right().get());
  //       if (field_expr->is_correlated()) {
  //         field_expr->set_tuple(tuple);
  //       }
  //     }
  //   }
  // }

  RC rc = physical_get_value(value);
  if (rc != RC::SUCCESS) {
    return rc;
  }
  return RC::SUCCESS;
}

const std::unique_ptr<SelectSqlNode> &SubQueryExpr::get_sql_node() const { return sql_node_; }

void SubQueryExpr::set_select_stmt(SelectStmt *select_stmt) { select_stmt_ = std::unique_ptr<SelectStmt>(select_stmt); }

const std::unique_ptr<SelectStmt> &SubQueryExpr::get_select_stmt() const { return select_stmt_; }

void SubQueryExpr::set_logical_operator(std::unique_ptr<LogicalOperator> &&logical_operator)
{
  logical_operator_ = std::move(logical_operator);
}

const std::unique_ptr<LogicalOperator> &SubQueryExpr::get_logical_operator() const { return logical_operator_; }

void SubQueryExpr::set_physical_operator(std::unique_ptr<PhysicalOperator> &&physical_operator)
{
  physical_operator_ = std::move(physical_operator);
}

const std::unique_ptr<PhysicalOperator> &SubQueryExpr::get_physical_operator() const { return physical_operator_; }

/////////////////////////////////////////////////////////////////////////////////
CastExpr::CastExpr(unique_ptr<Expression> child, AttrType cast_type) : child_(std::move(child)), cast_type_(cast_type)
{}

CastExpr::~CastExpr() {}

RC CastExpr::cast(const Value &value, Value &cast_value) const
{
  RC rc = RC::SUCCESS;
  if (this->value_type() == value.attr_type()) {
    cast_value = value;
    return rc;
  }
  rc = Value::cast_to(value, cast_type_, cast_value);
  return rc;
}

RC CastExpr::get_value(const Tuple &tuple, Value &result) const
{
  Value value;
  RC    rc = child_->get_value(tuple, value);
  if (rc != RC::SUCCESS) {
    return rc;
  }

  return cast(value, result);
}

RC CastExpr::try_get_value(Value &result) const
{
  Value value;
  RC    rc = child_->try_get_value(value);
  if (rc != RC::SUCCESS) {
    return rc;
  }

  return cast(value, result);
}

////////////////////////////////////////////////////////////////////////////////

ComparisonExpr::ComparisonExpr(CompOp comp, unique_ptr<Expression> left, unique_ptr<Expression> right)
    : comp_(comp), left_(std::move(left)), right_(std::move(right))
{}

ComparisonExpr::~ComparisonExpr() {}

RC ComparisonExpr::compare_value(const Value &left, const Value &right, bool &result) const
{
  RC rc = RC::SUCCESS;
  // 如果是NULL的判断，在这里做，不使用compare函数
  if (left.is_null() || right.is_null()) {
    switch (comp_) {
      case IS_NOT_NULL: {
        result = !(left.is_null() && right.is_null());
      } break;
      case IS_NULL: {
        result = left.is_null() && right.is_null();
      } break;
      default: {
        result = false;
      }
    }
    return rc;
  }
  int cmp_result = left.compare(right);
  result         = false;
  switch (comp_) {
    case EQUAL_TO: {
      result = (0 == cmp_result);
    } break;
    case LESS_EQUAL: {
      result = (cmp_result <= 0);
    } break;
    case NOT_EQUAL: {
      result = (cmp_result != 0);
    } break;
    case LESS_THAN: {
      result = (cmp_result < 0);
    } break;
    case GREAT_EQUAL: {
      result = (cmp_result >= 0);
    } break;
    case GREAT_THAN: {
      result = (cmp_result > 0);
    } break;
    case LIKE_WITH: {
      // TODO 需要填充如何来比较，其实上边可以不必管他
      result = left.is_like(right);
    } break;
    case NOT_LIKE_WITH: {
      // TODO 需要填充如何来比较，其实上边可以不必管他
      result = !left.is_like(right);
    } break;
    case IS_NULL:
    case IS_NOT_NULL:
    default: {
      LOG_WARN("unsupported comparison. %d", comp_);
      rc = RC::INTERNAL;
    } break;
  }

  return rc;
}

RC ComparisonExpr::try_get_value(Value &cell) const
{
  if (left_->type() == ExprType::VALUE && right_->type() == ExprType::VALUE) {
    ValueExpr   *left_value_expr  = static_cast<ValueExpr *>(left_.get());
    ValueExpr   *right_value_expr = static_cast<ValueExpr *>(right_.get());
    const Value &left_cell        = left_value_expr->get_value();
    const Value &right_cell       = right_value_expr->get_value();

    bool value = false;
    RC   rc    = compare_value(left_cell, right_cell, value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to compare tuple cells. rc=%s", strrc(rc));
    } else {
      cell.set_boolean(value);
    }
    return rc;
  }

  return RC::INVALID_ARGUMENT;
}

/**
 * 实际上是分别取得了左右值
 * 然后就比较两个值的结果是否相等
 * 然后返回值就设置为value，在这里实际上是一个bool值，true表示符合条件，false表示不符合条件
 */
RC ComparisonExpr::get_value(const Tuple &tuple, Value &value) const
{
  std::function<RC(SubQueryExpr *, const Tuple &)> set_correlated_field_expr;
  set_correlated_field_expr = [&](SubQueryExpr *sub_query_expr, const Tuple &tuple) -> RC {
    PhysicalOperator *phy_opr = sub_query_expr->get_physical_operator().get();
    // 这里应该逐层往下找FieldExpr
    Tuple *tuple_tmp = &*const_cast<Tuple *>(&tuple);
    LOG_TRACE("correlated sub query detected!");
    // 这里取得的应该是groupby physical operator
    if (phy_opr->children().front()->type() == PhysicalOperatorType::SCALAR_GROUP_BY) {
      vector<unique_ptr<PhysicalOperator>> &group_by_phy_opr = phy_opr->children();
      if (group_by_phy_opr.empty()) {
        LOG_WARN("no group_by physical operator");
        return RC::INTERNAL;
      }
      phy_opr = group_by_phy_opr[0].get();
    }
    vector<unique_ptr<PhysicalOperator>> &predicate_phy_opr = phy_opr->children();
    if (predicate_phy_opr.empty()) {
      LOG_WARN("no predicate physical operator");
      return RC::INTERNAL;
    }
    PredicatePhysicalOperator *predicate_operator =
        static_cast<PredicatePhysicalOperator *>(predicate_phy_opr[0].get());
    if (static_cast<PhysicalOperator *>(predicate_operator)->type() == PhysicalOperatorType::PREDICATE &&
        predicate_operator->expression().get() != nullptr) {

      unique_ptr<Expression> &conjunction_exprs = predicate_operator->expression();
      // conjunction_expr 可能有多个ConjunctionExpr
      for (auto &conjunction_expr : static_cast<ConjunctionExpr *>(conjunction_exprs.get())->children()) {
        ConjunctionExpr *expr_tmp = static_cast<ConjunctionExpr *>(conjunction_expr.get());
        if (!expr_tmp->children().empty()) {
          for (auto &comp_expr : expr_tmp->children()) {
            ComparisonExpr *comparison_expr = static_cast<ComparisonExpr *>(comp_expr.get());
            if (comparison_expr == nullptr || comparison_expr->left().get() == nullptr ||
                comparison_expr->right().get() == nullptr) {
              break;
            }
            if (comparison_expr->left()->type() == ExprType::FIELD) {
              FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->left().get());
              if (field_expr->is_correlated()) {
                field_expr->add_tuple(tuple_tmp);
              }
            } else if (comparison_expr->left()->type() == ExprType::SUB_QUERY) {
              SubQueryExpr *sub_query_expr = static_cast<SubQueryExpr *>(comparison_expr->left().get());
              set_correlated_field_expr(sub_query_expr, tuple);
            } else {
              return RC::INTERNAL;
            }
            if (comparison_expr->right()->type() == ExprType::FIELD) {
              FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->right().get());
              if (field_expr->is_correlated()) {
                field_expr->add_tuple(tuple_tmp);
              }
            } else if (comparison_expr->right()->type() == ExprType::SUB_QUERY) {
              SubQueryExpr *sub_query_expr = static_cast<SubQueryExpr *>(comparison_expr->right().get());
              set_correlated_field_expr(sub_query_expr, tuple);
            } else {
              return RC::INTERNAL;
            }
          }
        }
      }
    }
    return RC::SUCCESS;
  };
  std::function<RC(SubQueryExpr *)> delete_correlated_field_expr;
  delete_correlated_field_expr = [&](SubQueryExpr *sub_query_expr) -> RC {
    PhysicalOperator *phy_opr = sub_query_expr->get_physical_operator().get();
    // 这里应该逐层往下找FieldExpr
    LOG_TRACE("correlated sub query detected!");
    // 这里取得的应该是groupby physical operator
    if (phy_opr->children().front()->type() == PhysicalOperatorType::SCALAR_GROUP_BY) {
      vector<unique_ptr<PhysicalOperator>> &group_by_phy_opr = phy_opr->children();
      if (group_by_phy_opr.empty()) {
        LOG_WARN("no group_by physical operator");
        return RC::INTERNAL;
      }
      phy_opr = group_by_phy_opr[0].get();
    }
    vector<unique_ptr<PhysicalOperator>> &predicate_phy_opr = phy_opr->children();
    if (predicate_phy_opr.empty()) {
      LOG_WARN("no predicate physical operator");
      return RC::INTERNAL;
    }
    PredicatePhysicalOperator *predicate_operator =
        static_cast<PredicatePhysicalOperator *>(predicate_phy_opr[0].get());
    if (static_cast<PhysicalOperator *>(predicate_operator)->type() == PhysicalOperatorType::PREDICATE &&
        predicate_operator->expression().get() != nullptr) {

      unique_ptr<Expression> &conjunction_exprs = predicate_operator->expression();
      // conjunction_expr 可能有多个ComparisonExpr
      for (auto &conjunction_expr : static_cast<ConjunctionExpr *>(conjunction_exprs.get())->children()) {
        ConjunctionExpr *expr_tmp = static_cast<ConjunctionExpr *>(conjunction_expr.get());
        if (!expr_tmp->children().empty()) {
          for (auto &comp_expr : expr_tmp->children()) {
            ComparisonExpr *comparison_expr = static_cast<ComparisonExpr *>(comp_expr.get());
            if (comparison_expr == nullptr || comparison_expr->left().get() == nullptr ||
                comparison_expr->right().get() == nullptr) {
              break;
            }
            if (comparison_expr->left()->type() == ExprType::FIELD) {
              FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->left().get());
              if (field_expr->is_correlated()) {
                field_expr->pop_tuple();
              }
            } else if (comparison_expr->left()->type() == ExprType::SUB_QUERY) {
              SubQueryExpr *sub_query_expr = static_cast<SubQueryExpr *>(comparison_expr->left().get());
              delete_correlated_field_expr(sub_query_expr);
            } else {
              return RC::INTERNAL;
            }
            if (comparison_expr->right()->type() == ExprType::FIELD) {
              FieldExpr *field_expr = static_cast<FieldExpr *>(comparison_expr->right().get());
              if (field_expr->is_correlated()) {
                field_expr->pop_tuple();
              }
            } else if (comparison_expr->right()->type() == ExprType::SUB_QUERY) {
              SubQueryExpr *sub_query_expr = static_cast<SubQueryExpr *>(comparison_expr->right().get());
              set_correlated_field_expr(sub_query_expr, tuple);
            } else {
              return RC::INTERNAL;
            }
          }
        }
      }
    }
    return RC::SUCCESS;
  };

  if (comp_ == EXISTS_OP || comp_ == NOT_EXISTS_OP) {
    // 这里是exists的情况，没有左值，只有右边一个子查询
    SubQueryExpr *sub_query_expr = static_cast<SubQueryExpr *>(right_.get());
    bool          flag           = sub_query_expr->has_tuple();
    if (comp_ == EXISTS_OP) {
      value.set_boolean(flag);
    } else {
      value.set_boolean(!flag);
    }
    return RC::SUCCESS;
  }

  // 判断右边是不是直接的元组
  if (right_is_tuple_values()) {
    Value value_left;
    Value value_right;
    left_->get_value(tuple, value_left);
    vector<Value> values_right;
    TupleExpr    *tuple_expr = static_cast<TupleExpr *>(right_.get());
    tuple_expr->get_colums_values(values_right);

    // 下面进行比较
    bool flag = false;
    for (auto &value_right : values_right) {
      // 左右都不是NULL并且左右值相等才算在里边
      if ((!value_left.is_null() && !value_right.is_null()) && value_left.compare(value_right) == 0) {
        flag = true;
        break;
      }
    }
    if (comp_ == IN_OP) {
      value.set_boolean(flag);
    } else if (comp_ == NOT_IN_OP) {
      value.set_boolean(!flag);
    } else {
      return RC::INVALID_ARGUMENT;
    }
    return RC::SUCCESS;
  }

  // 判断右边是不是子查询
  if (right_is_sub_query() && !left_is_sub_query()) {
    Value value_left;
    Value value_right;
    left_->get_value(tuple, value_left);
    vector<Value> values_right;
    SubQueryExpr *sub_query_expr = static_cast<SubQueryExpr *>(right_.get());

    set_correlated_field_expr(sub_query_expr, tuple);
    RC rc = sub_query_expr->get_column_values(values_right);
    delete_correlated_field_expr(sub_query_expr);

    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get column values from sub query");
      return rc;
    }
    // 首先检查是不是子查询IN/NOT IN
    if (comp_ == IN_OP || comp_ == NOT_IN_OP) {
      bool flag = false;
      for (auto &value_right : values_right) {
        // 左右都不是NULL并且左右值相等才算在里边
        if ((!value_left.is_null() && !value_right.is_null()) && value_left.compare(value_right) == 0) {
          flag = true;
          break;
        }
      }
      if (comp_ == IN_OP) {
        value.set_boolean(flag);
      } else {
        value.set_boolean(!flag);
      }
      return RC::SUCCESS;
    } else {
      // 这里就是一个属性值与子查询的比较了
      // 这里需要注意的是，子查询的值可能是NULL
      if (values_right.size() > 1) {
        LOG_WARN("sub query has more than one value");
        return RC::INVALID_ARGUMENT;
      }
      if (values_right.size() == 0) {
        LOG_TRACE("sub query has no value");
        value.set_boolean(false);
        return RC::SUCCESS;
      }
      bool flag = !value_left.is_null() && !values_right[0].is_null();
      if (!flag) {
        value.set_boolean(flag);
        return RC::SUCCESS;
      }
      rc = compare_value(value_left, values_right[0], flag);
      if (rc != RC::SUCCESS) {
        LOG_WARN("failed to compare value with sub query");
        return rc;
      }
      value.set_boolean(flag);
      return RC::SUCCESS;
    }
  }

  if (left_is_sub_query() && !right_is_sub_query()) {
    // 首先检查是不是子查询IN/NOT IN
    Value value_left;
    Value value_right;
    right_->get_value(tuple, value_right);
    vector<Value> values_left;
    SubQueryExpr *sub_query_expr = static_cast<SubQueryExpr *>(left_.get());

    set_correlated_field_expr(sub_query_expr, tuple);

    RC rc = sub_query_expr->get_column_values(values_left);
    delete_correlated_field_expr(sub_query_expr);

    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get column values from sub query");
      return rc;
    }
    if (comp_ == IN_OP || comp_ == NOT_IN_OP) {
      bool flag = false;
      for (auto &value_left : values_left) {
        // 左右都不是NULL并且左右值相等才算在里边
        if ((!value_left.is_null() && !value_right.is_null()) && value_left.compare(value_right) == 0) {
          flag = true;
          break;
        }
      }
      if (comp_ == IN_OP) {
        value.set_boolean(flag);
      } else {
        value.set_boolean(!flag);
      }
      return RC::SUCCESS;
    } else {
      // 这里就是一个属性值与子查询的比较了
      // 这里需要注意的是，子查询的值可能是NULL
      if (values_left.size() > 1) {
        LOG_WARN("sub query has more than one value");
        return RC::INVALID_ARGUMENT;
      }
      if (values_left.size() == 0) {
        LOG_TRACE("sub query has no value");
        value.set_boolean(false);
        return RC::SUCCESS;
      }
      bool flag = !value_right.is_null() && !values_left[0].is_null();
      if (!flag) {
        value.set_boolean(flag);
        return RC::SUCCESS;
      }
      rc = compare_value(values_left[0], value_right, flag);
      if (rc != RC::SUCCESS) {
        LOG_WARN("failed to compare value with sub query");
        return rc;
      }
      value.set_boolean(flag);
      return RC::SUCCESS;
    }
  }

  if (left_is_sub_query() && right_is_sub_query()) {
    // 先处理left
    Value         value_left;
    vector<Value> values_left;
    SubQueryExpr *sub_query_expr = static_cast<SubQueryExpr *>(left_.get());

    set_correlated_field_expr(sub_query_expr, tuple);

    RC rc = sub_query_expr->get_column_values(values_left);
    delete_correlated_field_expr(sub_query_expr);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get column values from sub query");
      return rc;
    }

    // 处理right
    Value         value_right;
    vector<Value> values_right;
    SubQueryExpr *right_sub_query_expr = static_cast<SubQueryExpr *>(right_.get());

    set_correlated_field_expr(right_sub_query_expr, tuple);

    rc = right_sub_query_expr->get_column_values(values_right);
    delete_correlated_field_expr(right_sub_query_expr);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get column values from sub query");
      return rc;
    }

    if (comp_ == IN_OP || comp_ == NOT_IN_OP) {
      if (values_left.size() > 1) {
        LOG_WARN("left sub query has more than one value");
        return RC::INVALID_ARGUMENT;
      } else if (values_left.size() == 0) {
        LOG_TRACE("left sub query has no value");
        if (comp_ == IN_OP) {
          value.set_boolean(false);
        } else {
          value.set_boolean(true);
        }
        return RC::SUCCESS;
      } else if (values_right.size() > 1) {
        LOG_WARN("right sub query has more than one value");
        return RC::INVALID_ARGUMENT;
      } else if (values_right.size() == 0) {
        LOG_TRACE("right sub query has no value");
        if (comp_ == IN_OP) {
          value.set_boolean(false);
        } else {
          value.set_boolean(true);
        }
        return RC::SUCCESS;
      } else {
        bool flag = false;
        // 左右都不是NULL并且左右值相等才算在里边
        if ((!values_left[0].is_null() && !values_right[0].is_null()) && values_left[0].compare(values_right[0]) == 0) {
          flag = true;
        }
        if (comp_ == IN_OP) {
          value.set_boolean(flag);
        } else {
          value.set_boolean(!flag);
        }
      }
      return RC::SUCCESS;
    } else {
      // 这里就是子查询与子查询的比较了
      // 这里需要注意的是，子查询的值可能是NULL
      if (values_left.size() > 1 || values_right.size() > 1) {
        LOG_WARN("sub query has more than one value");
        return RC::INVALID_ARGUMENT;
      }
      if (values_left.size() == 0 || values_right.size() == 0) {
        LOG_TRACE("sub query has no value");
        value.set_boolean(false);
        return RC::SUCCESS;
      }
      bool flag = !values_right[0].is_null() && !values_left[0].is_null();
      if (!flag) {
        value.set_boolean(flag);
        return RC::SUCCESS;
      }
      rc = compare_value(values_left[0], values_right[0], flag);
      if (rc != RC::SUCCESS) {
        LOG_WARN("failed to compare value with sub query");
        return rc;
      }
      value.set_boolean(flag);
      return RC::SUCCESS;
    }
  }

  Value left_value;
  Value right_value;

  RC rc = left_->get_value(tuple, left_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }
  rc = right_->get_value(tuple, right_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
    return rc;
  }

  bool bool_value = false;

  rc = compare_value(left_value, right_value, bool_value);
  if (rc == RC::SUCCESS) {
    value.set_boolean(bool_value);
  }
  return rc;
}

RC ComparisonExpr::eval(Chunk &chunk, std::vector<uint8_t> &select)
{
  RC     rc = RC::SUCCESS;
  Column left_column;
  Column right_column;

  rc = left_->get_column(chunk, left_column);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }
  rc = right_->get_column(chunk, right_column);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
    return rc;
  }
  if (left_column.attr_type() != right_column.attr_type()) {
    LOG_WARN("cannot compare columns with different types");
    return RC::INTERNAL;
  }
  if (left_column.attr_type() == AttrType::INTS) {
    rc = compare_column<int>(left_column, right_column, select);
  } else if (left_column.attr_type() == AttrType::FLOATS) {
    rc = compare_column<float>(left_column, right_column, select);
  } else {
    // TODO: support string compare
    LOG_WARN("unsupported data type %d", left_column.attr_type());
    return RC::INTERNAL;
  }
  return rc;
}

template <typename T>
RC ComparisonExpr::compare_column(const Column &left, const Column &right, std::vector<uint8_t> &result) const
{
  RC rc = RC::SUCCESS;

  bool left_const  = left.column_type() == Column::Type::CONSTANT_COLUMN;
  bool right_const = right.column_type() == Column::Type::CONSTANT_COLUMN;
  if (left_const && right_const) {
    compare_result<T, true, true>((T *)left.data(), (T *)right.data(), left.count(), result, comp_);
  } else if (left_const && !right_const) {
    compare_result<T, true, false>((T *)left.data(), (T *)right.data(), right.count(), result, comp_);
  } else if (!left_const && right_const) {
    compare_result<T, false, true>((T *)left.data(), (T *)right.data(), left.count(), result, comp_);
  } else {
    compare_result<T, false, false>((T *)left.data(), (T *)right.data(), left.count(), result, comp_);
  }
  return rc;
}

////////////////////////////////////////////////////////////////////////////////
ConjunctionExpr::ConjunctionExpr(Type type, vector<unique_ptr<Expression>> &children)
    : conjunction_type_(type), children_(std::move(children))
{}

RC ConjunctionExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = RC::SUCCESS;
  if (children_.empty()) {
    value.set_boolean(true);
    return rc;
  }

  Value tmp_value;
  for (const unique_ptr<Expression> &expr : children_) {
    rc = expr->get_value(tuple, tmp_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value by child expression. rc=%s", strrc(rc));
      return rc;
    }
    bool bool_value = tmp_value.get_boolean();
    if ((conjunction_type_ == Type::AND && !bool_value) || (conjunction_type_ == Type::OR && bool_value)) {
      value.set_boolean(bool_value);
      return rc;
    }
  }

  bool default_value = (conjunction_type_ == Type::AND);
  value.set_boolean(default_value);
  return rc;
}

////////////////////////////////////////////////////////////////////////////////

ArithmeticExpr::ArithmeticExpr(ArithmeticExpr::Type type, Expression *left, Expression *right)
    : arithmetic_type_(type), left_(left), right_(right)
{}
ArithmeticExpr::ArithmeticExpr(ArithmeticExpr::Type type, unique_ptr<Expression> left, unique_ptr<Expression> right)
    : arithmetic_type_(type), left_(std::move(left)), right_(std::move(right))
{}

bool ArithmeticExpr::equal(const Expression &other) const
{
  if (this == &other) {
    return true;
  }
  if (type() != other.type()) {
    return false;
  }
  auto &other_arith_expr = static_cast<const ArithmeticExpr &>(other);
  return arithmetic_type_ == other_arith_expr.arithmetic_type() && left_->equal(*other_arith_expr.left_) &&
         right_->equal(*other_arith_expr.right_);
}
AttrType ArithmeticExpr::value_type() const
{
  if (!right_) {
    return left_->value_type();
  }

  if (left_->value_type() == AttrType::INTS && right_->value_type() == AttrType::INTS &&
      arithmetic_type_ != Type::DIV) {
    return AttrType::INTS;
  }

  return AttrType::FLOATS;
}

RC ArithmeticExpr::calc_value(const Value &left_value, const Value &right_value, Value &value) const
{
  RC rc = RC::SUCCESS;

  const AttrType target_type = value_type();
  value.set_type(target_type);

  switch (arithmetic_type_) {
    case Type::ADD: {
      Value::add(left_value, right_value, value);
    } break;

    case Type::SUB: {
      Value::subtract(left_value, right_value, value);
    } break;

    case Type::MUL: {
      Value::multiply(left_value, right_value, value);
    } break;

    case Type::DIV: {
      Value::divide(left_value, right_value, value);
    } break;

    case Type::NEGATIVE: {
      Value::negative(left_value, value);
    } break;

    default: {
      rc = RC::INTERNAL;
      LOG_WARN("unsupported arithmetic type. %d", arithmetic_type_);
    } break;
  }
  return rc;
}

template <bool LEFT_CONSTANT, bool RIGHT_CONSTANT>
RC ArithmeticExpr::execute_calc(
    const Column &left, const Column &right, Column &result, Type type, AttrType attr_type) const
{
  RC rc = RC::SUCCESS;
  switch (type) {
    case Type::ADD: {
      if (attr_type == AttrType::INTS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, int, AddOperator>(
            (int *)left.data(), (int *)right.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, AddOperator>(
            (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
    } break;
    case Type::SUB:
      if (attr_type == AttrType::INTS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, int, SubtractOperator>(
            (int *)left.data(), (int *)right.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, SubtractOperator>(
            (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
      break;
    case Type::MUL:
      if (attr_type == AttrType::INTS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, int, MultiplyOperator>(
            (int *)left.data(), (int *)right.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, MultiplyOperator>(
            (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
      break;
    case Type::DIV:
      if (attr_type == AttrType::INTS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, int, DivideOperator>(
            (int *)left.data(), (int *)right.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, DivideOperator>(
            (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
      break;
    case Type::NEGATIVE:
      if (attr_type == AttrType::INTS) {
        unary_operator<LEFT_CONSTANT, int, NegateOperator>((int *)left.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        unary_operator<LEFT_CONSTANT, float, NegateOperator>(
            (float *)left.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
      break;
    default: rc = RC::UNIMPLEMENTED; break;
  }
  if (rc == RC::SUCCESS) {
    result.set_count(result.capacity());
  }
  return rc;
}

RC ArithmeticExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = RC::SUCCESS;

  Value left_value;
  Value right_value;

  rc = left_->get_value(tuple, left_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }
  if (right_ == nullptr) {
    right_value = Value(0);
  } else {
    rc = right_->get_value(tuple, right_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
      return rc;
    }
  }
  return calc_value(left_value, right_value, value);
}

RC ArithmeticExpr::get_column(Chunk &chunk, Column &column)
{
  RC rc = RC::SUCCESS;
  if (pos_ != -1) {
    column.reference(chunk.column(pos_));
    return rc;
  }
  Column left_column;
  Column right_column;

  rc = left_->get_column(chunk, left_column);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get column of left expression. rc=%s", strrc(rc));
    return rc;
  }
  rc = right_->get_column(chunk, right_column);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get column of right expression. rc=%s", strrc(rc));
    return rc;
  }
  return calc_column(left_column, right_column, column);
}

RC ArithmeticExpr::calc_column(const Column &left_column, const Column &right_column, Column &column) const
{
  RC rc = RC::SUCCESS;

  const AttrType target_type = value_type();
  column.init(target_type, left_column.attr_len(), std::max(left_column.count(), right_column.count()));
  bool left_const  = left_column.column_type() == Column::Type::CONSTANT_COLUMN;
  bool right_const = right_column.column_type() == Column::Type::CONSTANT_COLUMN;
  if (left_const && right_const) {
    column.set_column_type(Column::Type::CONSTANT_COLUMN);
    rc = execute_calc<true, true>(left_column, right_column, column, arithmetic_type_, target_type);
  } else if (left_const && !right_const) {
    column.set_column_type(Column::Type::NORMAL_COLUMN);
    rc = execute_calc<true, false>(left_column, right_column, column, arithmetic_type_, target_type);
  } else if (!left_const && right_const) {
    column.set_column_type(Column::Type::NORMAL_COLUMN);
    rc = execute_calc<false, true>(left_column, right_column, column, arithmetic_type_, target_type);
  } else {
    column.set_column_type(Column::Type::NORMAL_COLUMN);
    rc = execute_calc<false, false>(left_column, right_column, column, arithmetic_type_, target_type);
  }
  return rc;
}

RC ArithmeticExpr::try_get_value(Value &value) const
{
  RC rc = RC::SUCCESS;

  Value left_value;
  Value right_value;

  rc = left_->try_get_value(left_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }

  if (right_) {
    rc = right_->try_get_value(right_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
      return rc;
    }
  }

  return calc_value(left_value, right_value, value);
}

////////////////////////////////////////////////////////////////////////////////

UnboundAggregateExpr::UnboundAggregateExpr(const char *aggregate_name, Expression *child)
    : aggregate_name_(aggregate_name), child_(child)
{}

////////////////////////////////////////////////////////////////////////////////
AggregateExpr::AggregateExpr(Type type, Expression *child) : aggregate_type_(type), child_(child) {}

AggregateExpr::AggregateExpr(Type type, unique_ptr<Expression> child) : aggregate_type_(type), child_(std::move(child))
{}

RC AggregateExpr::get_column(Chunk &chunk, Column &column)
{
  RC rc = RC::SUCCESS;
  if (pos_ != -1) {
    column.reference(chunk.column(pos_));
  } else {
    rc = RC::INTERNAL;
  }
  return rc;
}

bool AggregateExpr::equal(const Expression &other) const
{
  if (this == &other) {
    return true;
  }
  if (other.type() != type()) {
    return false;
  }
  const AggregateExpr &other_aggr_expr = static_cast<const AggregateExpr &>(other);
  return aggregate_type_ == other_aggr_expr.aggregate_type() && child_->equal(*other_aggr_expr.child());
}

unique_ptr<Aggregator> AggregateExpr::create_aggregator() const
{
  unique_ptr<Aggregator> aggregator;
  switch (aggregate_type_) {
    case Type::SUM: {
      aggregator = make_unique<SumAggregator>();
      break;
    }
    case Type::COUNT: {
      aggregator = make_unique<CountAggregator>();
      break;
    }
    case Type::AVG: {
      aggregator = make_unique<AvgAggregator>();
      break;
    }
    case Type::MAX: {
      aggregator = make_unique<MaxAggregator>();
      break;
    }
    case Type::MIN: {
      aggregator = make_unique<MinAggregator>();
      break;
    }
    default: {
      ASSERT(false, "unsupported aggregate type");
      break;
    }
  }
  return aggregator;
}

RC AggregateExpr::get_value(const Tuple &tuple, Value &value) const
{
  return tuple.find_cell(TupleCellSpec(name()), value);
}

RC AggregateExpr::type_from_string(const char *type_str, AggregateExpr::Type &type)
{
  RC rc = RC::SUCCESS;
  if (0 == strcasecmp(type_str, "count")) {
    type = Type::COUNT;
  } else if (0 == strcasecmp(type_str, "sum")) {
    type = Type::SUM;
  } else if (0 == strcasecmp(type_str, "avg")) {
    type = Type::AVG;
  } else if (0 == strcasecmp(type_str, "max")) {
    type = Type::MAX;
  } else if (0 == strcasecmp(type_str, "min")) {
    type = Type::MIN;
  } else {
    rc = RC::INVALID_ARGUMENT;
  }
  return rc;
}

void TupleExpr::get_colums_values(vector<Value> &values)
{
  for (unsigned i = 0; i < values_.size(); i++) {
    values.push_back(*values_[i]);
  }
}

RC LengthExpr::get_value(const Tuple &tuple, Value &value) const
{
  Value str_value;
  RC    rc = child_->get_value(tuple, str_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of child expression. rc=%s", strrc(rc));
    return rc;
  }
  if (str_value.is_null()) {
    value.set_null_flag(true);
    return RC::SUCCESS;
  }
  if (str_value.attr_type() != AttrType::CHARS) {
    LOG_WARN("length function requires a string type");
    return RC::INVALID_ARGUMENT;
  }
  int length = str_value.get_string().length();
  value      = Value(length);
  return RC::SUCCESS;
}

RC LengthExpr::try_get_value(Value &value) const
{
  Value str_value;
  RC    rc = child_->try_get_value(str_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of child expression. rc=%s", strrc(rc));
    return rc;
  }
  if (str_value.is_null()) {
    value.set_null_flag(true);
    return RC::SUCCESS;
  }
  if (str_value.attr_type() != AttrType::CHARS) {
    LOG_WARN("length function requires a string type");
    return RC::INVALID_ARGUMENT;
  }
  int length = str_value.get_string().length();
  value      = Value(length);
  return RC::SUCCESS;
}

RC RoundExpr::get_value(const Tuple &tuple, Value &value) const
{
  Value num_value;
  RC    rc = child_->get_value(tuple, num_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of child expression. rc=%s", strrc(rc));
    return rc;
  }
  if (num_value.is_null()) {
    value.set_null_flag(true);
    return RC::SUCCESS;
  }
  if (num_value.attr_type() != AttrType::INTS && num_value.attr_type() != AttrType::FLOATS) {
    LOG_WARN("round function requires a numeric type");
    return RC::INVALID_ARGUMENT;
  }
  if (num_value.attr_type() == AttrType::INTS) {
    value = num_value;
    return RC::SUCCESS;
  }
  double num = static_cast<double>(num_value.get_float());
  double result;

  if (decimals_ >= 0) {
    double factor = std::pow(10.0, decimals_);
    result        = std::round(num * factor + 1e-8) / factor;
  } else {
    // 负数 decimals_ 表示向左 round，比如 ROUND(1234, -2) = 1200
    double factor = std::pow(10.0, -decimals_);
    result        = std::round(num / factor + 1e-8) * factor;
  }

  value = Value(static_cast<float>(result));  // 存回 Value，保持原有 float 类型
  return RC::SUCCESS;
}

RC RoundExpr::try_get_value(Value &value) const
{
  Value num_value;
  RC    rc = child_->try_get_value(num_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of child expression. rc=%s", strrc(rc));
    return rc;
  }
  if (num_value.is_null()) {
    value.set_null_flag(true);
    return RC::SUCCESS;
  }
  if (num_value.attr_type() != AttrType::INTS && num_value.attr_type() != AttrType::FLOATS) {
    LOG_WARN("round function requires a numeric type");
    return RC::INVALID_ARGUMENT;
  }
  if (num_value.attr_type() == AttrType::INTS) {
    value = num_value;
    return RC::SUCCESS;
  }
  // 需要考虑decimals，这是要保留的小数的位数
  double num = static_cast<double>(num_value.get_float());
  double result;

  if (decimals_ >= 0) {
    double factor = std::pow(10.0, decimals_);
    result        = std::round(num * factor + 1e-8) / factor;
  } else {
    // 负数 decimals_ 表示向左 round，比如 ROUND(1234, -2) = 1200
    double factor = std::pow(10.0, -decimals_);
    result        = std::round(num / factor + 1e-8) * factor;
  }

  value = Value(static_cast<float>(result));  // 存回 Value，保持原有 float 类型
  return RC::SUCCESS;
}

RC DateFormatExpr::get_value(const Tuple &tuple, Value &value) const
{
  Value value_tmp;
  RC    rc = child_->get_value(tuple, value_tmp);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of child expression. rc=%s", strrc(rc));
    return rc;
  }
  if (value_tmp.is_null()) {
    value.set_null_flag(true);
    return RC::SUCCESS;
  }
  if (value_tmp.attr_type() != AttrType::DATES) {
    LOG_WARN("date format function requires a date type");
    return RC::INVALID_ARGUMENT;
  }
  int    date_int = value_tmp.get_int();
  string year_str = to_string((date_int / 10000) % 100);
  string Year_str = to_string(date_int / 10000);

  string month_str = to_string((date_int % 10000) / 100);
  string Month_str;
  switch ((date_int % 10000) / 100) {
    case 1: Month_str = "January"; break;
    case 2: Month_str = "February"; break;
    case 3: Month_str = "March"; break;
    case 4: Month_str = "April"; break;
    case 5: Month_str = "May"; break;
    case 6: Month_str = "June"; break;
    case 7: Month_str = "July"; break;
    case 8: Month_str = "August"; break;
    case 9: Month_str = "September"; break;
    case 10: Month_str = "October"; break;
    case 11: Month_str = "November"; break;
    case 12: Month_str = "December"; break;
    default: Month_str = month_str; break;
  }

  string day_str = to_string(date_int % 100);
  string Day_str;

  switch (date_int % 100) {
    case 1:
    case 21:
    case 31: Day_str = day_str + "st"; break;
    case 2:
    case 22: Day_str = day_str + "nd"; break;
    case 3:
    case 23: Day_str = day_str + "rd"; break;
    default: Day_str = day_str + "th"; break;
  }

  if (year_str.size() == 1) {
    year_str = "0" + year_str;
  }
  if (month_str.size() == 1) {
    month_str = "0" + month_str;
  }
  if (day_str.size() == 1) {
    day_str = "0" + day_str;
  }
  // format_是一个字符串，它的格式是%y...%m...%d之类的，应该把%year, %month, %day替换掉
  string date_format_str;
  // 不从0开始是因为format_的第一个字符是'，最后一个字符是'，不需要处理
  for (unsigned i = 1; i < format_.size() - 1; i++) {
    if (format_[i] == '%' && i + 1 < format_.size()) {
      char next_char = format_[i + 1];
      if (next_char == 'Y' || next_char == 'y') {
        if (next_char == 'Y') {
          date_format_str.append(Year_str);
        } else {
          date_format_str.append(year_str);
        }
        i++;
      } else if (next_char == 'M' || next_char == 'm') {
        if (next_char == 'M') {
          date_format_str.append(Month_str);
        } else {
          date_format_str.append(month_str);
        }
        i++;
      } else if (next_char == 'D' || next_char == 'd') {
        if (next_char == 'D') {
          date_format_str.append(Day_str);
        } else {
          date_format_str.append(day_str);
        }
        i++;
      } else {
        date_format_str.push_back(format_[i + 1]);
        i++;
      }
    } else {
      date_format_str.push_back(format_[i]);
    }
  }
  value = Value(date_format_str.c_str(), date_format_str.size());
  return RC::SUCCESS;
}

RC DateFormatExpr::try_get_value(Value &value) const
{
  Value value_tmp;
  RC    rc = child_->try_get_value(value_tmp);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of child expression. rc=%s", strrc(rc));
    return rc;
  }
  if (value_tmp.is_null()) {
    value.set_null_flag(true);
    return RC::SUCCESS;
  }
  if (value_tmp.attr_type() != AttrType::DATES) {
    LOG_WARN("date format function requires a date type");
    return RC::INVALID_ARGUMENT;
  }
  int    date_int = value_tmp.get_int();
  string year_str = to_string((date_int / 10000) % 100);
  string Year_str = to_string(date_int / 10000);

  string month_str = to_string((date_int % 10000) / 100);
  string Month_str;
  switch ((date_int % 10000) / 100) {
    case 1: Month_str = "January"; break;
    case 2: Month_str = "February"; break;
    case 3: Month_str = "March"; break;
    case 4: Month_str = "April"; break;
    case 5: Month_str = "May"; break;
    case 6: Month_str = "June"; break;
    case 7: Month_str = "July"; break;
    case 8: Month_str = "August"; break;
    case 9: Month_str = "September"; break;
    case 10: Month_str = "October"; break;
    case 11: Month_str = "November"; break;
    case 12: Month_str = "December"; break;
    default: Month_str = month_str; break;
  }

  string day_str = to_string(date_int % 100);
  string Day_str;

  switch (date_int % 100) {
    case 1:
    case 21:
    case 31: Day_str = day_str + "st"; break;
    case 2:
    case 22: Day_str = day_str + "nd"; break;
    case 3:
    case 23: Day_str = day_str + "rd"; break;
    default: Day_str = day_str + "th"; break;
  }
  if (year_str.size() == 1) {
    year_str = "0" + year_str;
  }
  if (month_str.size() == 1) {
    month_str = "0" + month_str;
  }
  if (day_str.size() == 1) {
    day_str = "0" + day_str;
  }
  // format_是一个字符串，它的格式是%y...%m...%d之类的，应该把%year, %month, %day替换掉
  string date_format_str;
  // 不从0开始是因为format_的第一个字符是'，最后一个字符是'，不需要处理
  for (unsigned i = 1; i < format_.size() - 1; i++) {
    if (format_[i] == '%' && i + 1 < format_.size() - 1) {
      char next_char = format_[i + 1];
      if (next_char == 'Y' || next_char == 'y') {
        if (next_char == 'Y') {
          date_format_str.append(Year_str);
        } else {
          date_format_str.append(year_str);
        }
        i++;
      } else if (next_char == 'M' || next_char == 'm') {
        if (next_char == 'M') {
          date_format_str.append(Month_str);
        } else {
          date_format_str.append(month_str);
        }
        i++;
      } else if (next_char == 'D' || next_char == 'd') {
        if (next_char == 'D') {
          date_format_str.append(Day_str);
        } else {
          date_format_str.append(day_str);
        }
        i++;
      } else {
        date_format_str.push_back(format_[i + 1]);
        i++;
      }
    } else {
      date_format_str.push_back(format_[i]);
    }
  }
  value = Value(date_format_str.c_str(), date_format_str.size());
  return RC::SUCCESS;
}
